// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */


#ifndef CEPH_FILER_H
#define CEPH_FILER_H

 /*** Filer
  *
  * stripe file ranges onto objects.
  * build list<ObjectExtent> for the objecter or objectcacher.
  *
  * also, provide convenience methods that call objecter for you.
  *
  * "files" are identified by ino.
  */


#include <mutex>

#include "include/types.h"

#include "common/ceph_time.h"

#include "osd/OSDMap.h"
#include "Objecter.h"
#include "Striper.h"

class Context;
class Messenger;
class OSDMap;
class Finisher;


/**** Filer interface ***/

class Filer {
	CephContext* cct;
	Objecter* objecter;
	Finisher* finisher;

	// probes
	struct Probe {
		std::mutex lock;
		typedef std::lock_guard<std::mutex> lock_guard;
		typedef std::unique_lock<std::mutex> unique_lock;
		inodeno_t ino;
		file_layout_t layout;
		snapid_t snapid;

		uint64_t* psize;
		ceph::real_time* pmtime;
		utime_t* pumtime;

		int flags;

		bool fwd;

		Context* onfinish;

		std::vector<ObjectExtent> probing;
		uint64_t probing_off, probing_len;

		std::map<object_t, uint64_t> known_size;
		ceph::real_time max_mtime;

		std::set<object_t> ops;

		int err;
		bool found_size;

		Probe(inodeno_t i, const file_layout_t& l, snapid_t sn,
			uint64_t f, uint64_t* e, ceph::real_time* m, int fl, bool fw,
			Context* c) :
			ino(i), layout(l), snapid(sn),
			psize(e), pmtime(m), pumtime(nullptr), flags(fl), fwd(fw), onfinish(c),
			probing_off(f), probing_len(0),
			err(0), found_size(false) {}

		Probe(inodeno_t i, const file_layout_t& l, snapid_t sn,
			uint64_t f, uint64_t* e, utime_t* m, int fl, bool fw,
			Context* c) :
			ino(i), layout(l), snapid(sn),
			psize(e), pmtime(nullptr), pumtime(m), flags(fl), fwd(fw),
			onfinish(c), probing_off(f), probing_len(0),
			err(0), found_size(false) {}
	};

	class C_Probe;

	void _probe(Probe* p, Probe::unique_lock& pl);
	bool _probed(Probe* p, const object_t& oid, uint64_t size,
		ceph::real_time mtime, Probe::unique_lock& pl);

public:
	Filer(const Filer& other);
	const Filer operator=(const Filer& other);

	Filer(Objecter* o, Finisher* f) : cct(o->cct), objecter(o), finisher(f) {}
	~Filer() {}

	bool is_active() {
		return objecter->is_active(); // || (oc && oc->is_active());
	}


	/*** async file interface.  scatter/gather as needed. ***/

	void read(inodeno_t ino,
		const file_layout_t* layout,
		snapid_t snap,
		uint64_t offset,
		uint64_t len,
		ceph::buffer::list* bl,   // ptr to data
		int flags,
		Context* onfinish,
		int op_flags = 0) {
		ceph_assert(snap);  // (until there is a non-NOSNAP write)
		std::vector<ObjectExtent> extents;
		Striper::file_to_extents(cct, ino, layout, offset, len, 0, extents);
		objecter->sg_read(extents, snap, bl, flags, onfinish, op_flags);
	}

	void read_trunc(inodeno_t ino,
		const file_layout_t* layout,
		snapid_t snap,
		uint64_t offset,
		uint64_t len,
		ceph::buffer::list* bl, // ptr to data
		int flags,
		uint64_t truncate_size,
		__u32 truncate_seq,
		Context* onfinish,
		int op_flags = 0) {
		ceph_assert(snap);  // (until there is a non-NOSNAP write)
		std::vector<ObjectExtent> extents;
		Striper::file_to_extents(cct, ino, layout, offset, len, truncate_size,
			extents);
		objecter->sg_read_trunc(extents, snap, bl, flags,
			truncate_size, truncate_seq, onfinish, op_flags);
	}

	void write(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t offset,
		uint64_t len,
		ceph::buffer::list& bl,
		ceph::real_time mtime,
		int flags,
		Context* oncommit,
		int op_flags = 0) {
		std::vector<ObjectExtent> extents;
		Striper::file_to_extents(cct, ino, layout, offset, len, 0, extents);
		objecter->sg_write(extents, snapc, bl, mtime, flags, oncommit, op_flags);
	}

	void write_trunc(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t offset,
		uint64_t len,
		ceph::buffer::list& bl,
		ceph::real_time mtime,
		int flags,
		uint64_t truncate_size,
		__u32 truncate_seq,
		Context* oncommit,
		int op_flags = 0) {
		std::vector<ObjectExtent> extents;
		Striper::file_to_extents(cct, ino, layout, offset, len, truncate_size,
			extents);
		objecter->sg_write_trunc(extents, snapc, bl, mtime, flags,
			truncate_size, truncate_seq, oncommit, op_flags);
	}

	void truncate(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t offset,
		uint64_t len,
		__u32 truncate_seq,
		ceph::real_time mtime,
		int flags,
		Context* oncommit);
	void _do_truncate_range(struct TruncRange* pr, int fin);

	void zero(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t offset,
		uint64_t len,
		ceph::real_time mtime,
		int flags,
		bool keep_first,
		Context* oncommit) {
		std::vector<ObjectExtent> extents;
		Striper::file_to_extents(cct, ino, layout, offset, len, 0, extents);
		if (extents.size() == 1) {
			if (extents[0].offset == 0 && extents[0].length == layout->object_size
				&& (!keep_first || extents[0].objectno != 0))
				objecter->remove(extents[0].oid, extents[0].oloc,
				snapc, mtime, flags, oncommit);
			else
				objecter->zero(extents[0].oid, extents[0].oloc, extents[0].offset,
				extents[0].length, snapc, mtime, flags, oncommit);
		} else {
			C_GatherBuilder gcom(cct, oncommit);
			for (auto p = extents.begin(); p != extents.end(); ++p) {
				if (p->offset == 0 && p->length == layout->object_size &&
					(!keep_first || p->objectno != 0))
					objecter->remove(p->oid, p->oloc,
					snapc, mtime, flags,
					oncommit ? gcom.new_sub() : 0);
				else
					objecter->zero(p->oid, p->oloc, p->offset, p->length,
					snapc, mtime, flags,
					oncommit ? gcom.new_sub() : 0);
			}
			gcom.activate();
		}
	}

	void zero(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t offset,
		uint64_t len,
		ceph::real_time mtime,
		int flags,
		Context* oncommit) {
		zero(ino, layout,
			snapc, offset,
			len, mtime,
			flags, false,
			oncommit);
	}
	// purge range of ino.### objects
	int purge_range(inodeno_t ino,
		const file_layout_t* layout,
		const SnapContext& snapc,
		uint64_t first_obj, uint64_t num_obj,
		ceph::real_time mtime,
		int flags, Context* oncommit);
	void _do_purge_range(struct PurgeRange* pr, int fin, int err);

	/*
	 * probe
	 *  specify direction,
	 *  and whether we stop when we find data, or hole.
	 */
	int probe(inodeno_t ino,
		const file_layout_t* layout,
		snapid_t snapid,
		uint64_t start_from,
		uint64_t* end,
		ceph::real_time* mtime,
		bool fwd,
		int flags,
		Context* onfinish);

	int probe(inodeno_t ino,
		const file_layout_t* layout,
		snapid_t snapid,
		uint64_t start_from,
		uint64_t* end,
		bool fwd,
		int flags,
		Context* onfinish) {
		return probe(ino, layout, snapid, start_from, end,
			(ceph::real_time*)0, fwd, flags, onfinish);
	}

	int probe(inodeno_t ino,
		const file_layout_t* layout,
		snapid_t snapid,
		uint64_t start_from,
		uint64_t* end,
		utime_t* mtime,
		bool fwd,
		int flags,
		Context* onfinish);

private:
	int probe_impl(Probe* probe, const file_layout_t* layout,
		uint64_t start_from, uint64_t* end);
};

#endif // !CEPH_FILER_H
